package atguigu.com.edu.app.dws;

import atguigu.com.edu.bean.CartAddUuBean;
import atguigu.com.edu.util.ClickHouseUtil;
import atguigu.com.edu.util.DateFormatUtil;
import atguigu.com.edu.util.MyClickhouseUtil;
import atguigu.com.edu.util.MyKafkaUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;


import java.time.Duration;


public class DwsTradeCartAddUuWindow {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);


       /* env.enableCheckpointing(60000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        env.getCheckpointConfig().setCheckpointTimeout(50000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop202:8020/edu/ck");
        System.setProperty("HADOOP_USER_NAME","atguigu");*/

       String topic = "dwd_trade_cart_add";
       String groupId = "dws_trade_cart_add_group";



        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> stringDataStreamSource = env.addSource(kafkaConsumer);

        //stringDataStreamSource.print("wwwww");

        SingleOutputStreamOperator<JSONObject> jsonObj = stringDataStreamSource.map(JSON::parseObject);

        SingleOutputStreamOperator<JSONObject> withWatermarkDS = jsonObj.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                                        return jsonObj.getLong("ts") * 1000;
                                    }
                                }
                        )
        );

        KeyedStream<JSONObject, String> keyedDs = withWatermarkDS.keyBy(r -> r.getString("user_id"));


        SingleOutputStreamOperator<JSONObject> uuDs = keyedDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

            private ValueState<String> last;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> lastV = new ValueStateDescriptor<>("last", String.class);

                lastV.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());

                last = getRuntimeContext().getState(lastV);


            }

            @Override
            public void processElement(JSONObject in, Context context, Collector<JSONObject> out) throws Exception {

                String value = last.value();

                Long ts = in.getLong("ts");
                String cur = DateFormatUtil.toDate(ts);


                if (StringUtils.isEmpty(value)||last.equals(cur)){

                    out.collect(in);
                    last.update(cur);
                }


            }
        });

        //开窗
        keyedDs.print("------------");
        AllWindowedStream<JSONObject, TimeWindow> window = uuDs.windowAll(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));


        SingleOutputStreamOperator<CartAddUuBean> aggW = window.aggregate(new AggregateFunction<JSONObject, Long, Long>() {
            @Override
            public Long createAccumulator() {
                return 0L;
            }

            @Override
            public Long add(JSONObject jsonObject, Long aLong) {
                return aLong++;
            }

            @Override
            public Long getResult(Long aLong) {
                return aLong;
            }

            @Override
            public Long merge(Long aLong, Long acc1) {
                return null;
            }
        }, new AllWindowFunction<Long, CartAddUuBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow timeWindow, Iterable<Long> iterable, Collector<CartAddUuBean> out) throws Exception {

                for (Long value : iterable) {
                    out.collect(new CartAddUuBean(
                            DateFormatUtil.toYmdHms(timeWindow.getStart()),
                            DateFormatUtil.toYmdHms(timeWindow.getEnd()),
                            value,
                            System.currentTimeMillis()


                    ));
                }


            }
        });



        aggW.print();

        aggW.addSink(ClickHouseUtil.getJdbcSink("insert into dws_trade_cart_add_uu_window values(?,?,?,?)"));

        env.execute();


    }
}
